package com.pw.gmall.realtime.function;

import com.pw.gmall.realtime.common.Constant;
import com.pw.gmall.realtime.utils.JdbcUtils;
import com.pw.gmall.realtime.utils.RedisUtils;
import com.pw.gmall.realtime.utils.ThreadPoolUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author: linux_future
 * @since: 2022/3/22
 **/
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {
    public abstract void addDim(Jedis jedis, Connection conn, T input, ResultFuture<T> resultFuture) throws Exception;
    private ThreadPoolExecutor pool;

    @Override
    public void open(Configuration parameters) throws Exception {
        pool = ThreadPoolUtils.getThreadPools();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        pool.execute(new Runnable() {

            private Jedis jedis;
            private Connection conn;

            @Override
            public void run() {
                //读取维度
                try {
                    conn = JdbcUtils.getJdbcConnect(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL, null, null);
                    jedis = RedisUtils.getClient();
                    //处理维度
                    addDim(jedis,conn,input,resultFuture);
                } catch (Exception e) {
                    e.printStackTrace();
                }finally {
                    //关闭连接对象
                    if(conn!=null){
                        try{
                            conn.close();
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                    }
                    if(jedis !=null){
                        jedis.close();
                    }
                }
            }


        });
    }
}
